package shuai7boy.vip.transform.mr.au;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;


import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.mapreduce.Reducer;

import shuai7boy.vip.common.DateEnum;
import shuai7boy.vip.common.KpiTypeEnum;
import shuai7boy.vip.transformer.model.dim.StatsUserDimension;
import shuai7boy.vip.transformer.model.value.map.TimeOutValue;
import shuai7boy.vip.transformer.model.value.reduce.MapWritableValue;
import shuai7boy.vip.util.TimeUtil;

public class ActiveUserReducer extends Reducer<StatsUserDimension, TimeOutValue, StatsUserDimension, MapWritableValue> {

	private Set<String> unique = new HashSet<String>();
	private MapWritableValue outputValue = new MapWritableValue();
	private MapWritable map = new MapWritable();

	private Map<Integer, HashSet<String>> hourlyUnique = new HashMap<Integer, HashSet<String>>();// key为浏览时间，value为浏览人
	private MapWritable hourlyMap=new MapWritable();
//	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		// 进行初始化操作
		for (int i = 0; i <24; i++) {
			this.hourlyMap.put(new IntWritable(i),new IntWritable(0));
			this.hourlyUnique.put(i,new HashSet<String>());
		}
	}

	@Override
	protected void reduce(StatsUserDimension key, Iterable<TimeOutValue> values, Context context)
			throws IOException, InterruptedException {

		try {
			// 添加时间维度计算
			String dimensionName = key.getStatsCommon().getKpi().getKpiName();
			if (dimensionName == KpiTypeEnum.HOURLY_ACTIVE_USER.name) {
				for(TimeOutValue tv:values) {					
					//计算访问的小时，从[0,23]的区段
					int hour=TimeUtil.getDateInfo(tv.getTime(), DateEnum.HOUR);					
					hourlyUnique.get(hour).add(tv.getId());
				}
				
				//设置kpi
				this.outputValue.setKpi(KpiTypeEnum.HOURLY_ACTIVE_USER);
				
				for(Map.Entry<Integer,HashSet<String>> entry:hourlyUnique.entrySet()) {
					hourlyMap.put(new IntWritable(entry.getKey()),new IntWritable(entry.getValue().size()));
				}
				
				this.outputValue.setValue(hourlyMap);
				context.write(key,this.outputValue);//输出值

			} else {
				// 将uuid添加到set集合中去，方便进行统计uuid的去重个数
				for (TimeOutValue value : values) {
					this.unique.add(value.getId());
				}
				// 设置kpi
				this.outputValue.setKpi(KpiTypeEnum.valueOfName(key.getStatsCommon().getKpi().getKpiName()));
				// 设置value
				this.map.put(new IntWritable(-1), new IntWritable(this.unique.size()));
				this.outputValue.setValue(this.map);

				// 进行输出
				context.write(key, this.outputValue);
			}

		} finally {
			// 清空操作
			this.unique.clear();
			this.hourlyUnique.clear();
			this.hourlyMap.clear();
			this.map.clear();
			for (int i = 0; i <24; i++) {
				this.hourlyMap.put(new IntWritable(i),new IntWritable(0));
				this.hourlyUnique.put(i,new HashSet<String>());
			}
		}

	}

}
